package com.atguigu.gmall.realtime.app.dwd.db;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2023/2/14 13:47
 */
public class Dwd_04_DwdTradeCancelDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_04_DwdTradeCancelDetail().init(
            3004,
            2,
            "Dwd_04_DwdTradeCancelDetail"
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));
        
        // 1. 读取 ods_db
        readOdsDb(tEnv, "Dwd_04_DwdTradeCancelDetail");
        // 2. 读取下单事务事实表 (上一个需求的结果)
        tEnv.executeSql("create table dwd_trade_order_detail( " +
                            "id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "sku_id string, " +
                            "sku_name string, " +
                            "province_id string, " +
                            "activity_id string, " +
                            "activity_rule_id string, " +
                            "coupon_id string, " +
                            "date_id string, " +
                            "create_time string, " +
                            "source_id string, " +
                            "source_type string, " +
                            "source_type_name string, " +
                            "sku_num string, " +
                            "split_original_amount string, " +
                            "split_activity_amount string, " +
                            "split_coupon_amount string, " +
                            "split_total_amount string, " +
                            "ts bigint " +
                            ")" + SQLUtil.getDDLKafkaSource(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_04_DwdTradeCancelDetail"));
        
        // 3. order_info 和下单事务事实表的 join: 内连接
        Table orderCancelInfo = tEnv.sqlQuery("select " +
                                                  "data['id'] id, " +
                                                  "data['user_id'] user_id, " +
                                                  "data['province_id'] province_id, " +
                                                  "data['operate_time'] operate_time, " +
                                                  "ts " +
                                                  "from ods_db " +
                                                  "where `database`='gmall2022' " +
                                                  "and `table`='order_info' " +
                                                  "and `type`='update' " +
                                                  "and `old`['order_status']='1001' " +
                                                  "and `data`['order_status']='1003' ");
        tEnv.createTemporaryView("order_cancel_info", orderCancelInfo);
        
        // 4. 结果写出到 kafka 中, 用普通的 kafka
        Table result = tEnv.sqlQuery("select " +
                                         "od.id, " +
                                         "od.order_id, " +
                                         "oci.user_id, " +
                                         "od.sku_id, " +
                                         "od.sku_name, " +
                                         "oci.province_id, " +
                                         "od.activity_id, " +
                                         "od.activity_rule_id, " +
                                         "od.coupon_id, " +
                                         "date_format(oci.operate_time, 'yyyy-MM-dd') operate_date_id, " +
                                         "oci.operate_time, " +
                                         "od.source_id, " +
                                         "od.source_type, " +
                                         "od.source_type_name, " +
                                         "od.sku_num, " +
                                         "od.split_original_amount, " +
                                         "od.split_activity_amount, " +
                                         "od.split_coupon_amount, " +
                                         "od.split_total_amount, " +
                                         "oci.ts " +
                                         "from order_cancel_info oci " +
                                         "join dwd_trade_order_detail od " +
                                         "on oci.id=od.order_id");
        
        tEnv.executeSql("create table dwd_trade_cancel_detail( " +
                            "id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "sku_id string, " +
                            "sku_name string, " +
                            "province_id string, " +
                            "activity_id string, " +
                            "activity_rule_id string, " +
                            "coupon_id string, " +
                            "date_id string, " +
                            "cancel_time string, " +
                            "source_id string, " +
                            "source_type_code string, " +
                            "source_type_name string, " +
                            "sku_num string, " +
                            "split_original_amount string, " +
                            "split_activity_amount string, " +
                            "split_coupon_amount string, " +
                            "split_total_amount string, " +
                            "ts bigint " +
                            ")" + SQLUtil.getDDLKafkaSink(Constant.TOPIC_DWD_TRADE_CANCEL_DETAIL));
    
        result.executeInsert("dwd_trade_cancel_detail");
        
    }
}
/*
交易域取消订单事务事实表

order_detail                insert(下单)
order_info                  update(取消)  order_status: 1001->1003
order_detail_activity       insert(下单)
order_detail_coupon         insert(下单)
base_dic

ttl:30min+5s

----
优化:
order_info                  update(取消)  order_status: 1001->1003

和 下单事务事实表进行 join

 */